package com.leyantech.ecp.agenx.session.manager.handler;

import static com.leyantech.ecp.agenx.common.config.ConfigManager.ASSISTANT_EVENT_TOPIC;
import static com.leyantech.ecp.agenx.common.utils.SplitterUtils.DOT_SPLITTER;

import com.leyantech.chaos.kafka.KafkaProducerStringClient;
import com.leyantech.ecp.agenx.common.handler.ActionHandler;
import com.leyantech.ecp.agenx.common.message.EnumWsFnType;
import com.leyantech.ecp.agenx.common.redis.RedissonOperation;
import com.leyantech.ecp.agenx.common.redis.RedissonOperation.RedisExecutor;
import com.leyantech.ecp.agenx.common.redis.entity.OnlineAssistantSession;
import com.leyantech.ecp.agenx.session.client.Assistant;
import com.leyantech.ecp.agenx.session.client.SessionClient;

import com.google.inject.Inject;
import com.google.inject.Singleton;
import io.vertx.core.Vertx;
import io.vertx.core.json.JsonObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.redisson.api.RBucket;
import org.redisson.api.RMap;
import org.redisson.api.RObject;
import org.redisson.api.RSet;

import java.time.Duration;
import java.util.List;
import java.util.concurrent.atomic.LongAdder;

/**
 * @author eahau, {@literal <eahau@foxmail.com>}
 * @date 2022-11-07.
 */
@Slf4j
@Singleton
public class AssistantLogoutHandler extends ActionHandler implements PathChildrenCacheListener {

  private final Vertx vertx;

  private final KafkaProducerStringClient kafkaClient;

  @Inject
  public AssistantLogoutHandler(Vertx vertx,
      KafkaProducerStringClient kafkaClient) {
    this.vertx = vertx;
    this.kafkaClient = kafkaClient;
  }

  @Override
  public EnumWsFnType actionType() {
    return EnumWsFnType.ASSISTANT_LOGOUT;
  }

  @Override
  public void handle(JsonObject message) {

    log.info("received assistant LOGOUT event {}", message);

    final JsonObject header = message.getJsonObject("header");

    final long timestamp = header.getLong("ts");

    final Assistant assistant = Assistant.fromJson(message.getJsonObject("body").toString());

    assistant.setTimestamp(timestamp);
    assistant.setAction(actionType().type);

    updateAssistantSession(assistant, true);

    sendMessage(assistant);

    // 客服登出，删除客服所在节点记录
    final String nodeId = assistant.getNodeId();
    RedissonOperation.NODE_ASSISTANT.executor(nodeId).run(set -> {
      final String key = String.join(".", assistant.getStoreId(), assistant.getAssistantId());
      if (set.remove(key)) {
        log.info("remove {} on node {} succeed", key, nodeId);
      }
    }, "remove");
  }

  private void updateAssistantSession(Assistant assistant, boolean ignoredNodeId) {
    final String storeId = assistant.getStoreId();
    final String assistantId = assistant.getAssistantId();

    final RedisExecutor<RMap<String, OnlineAssistantSession>> executor = RedissonOperation.STORE_ASSISTANT
        .executor(storeId);

    final OnlineAssistantSession assistantSession = executor
        .get(map -> map.get(assistantId), "get");

    if (assistantSession == null) {
      return;
    }

    if (assistant.getTimestamp() > assistantSession.getTimestamp()
        && (ignoredNodeId
        || StringUtils.equals(assistant.getNodeId(), assistantSession.getNodeId()))) {
      // remove
      executor.run(map -> {
        if (map.remove(assistantId) != null) {
          log.info("remove AssistantSession succeed, sid {} aid {}", storeId, assistantId);
        }
      }, "remove");
    }
  }

  /**
   * 通知其它服务客服登录登出.
   */
  private void sendMessage(Assistant assistant) {
    final String key = String.join(".", assistant.getStoreId(), assistant.getAssistantId());
    kafkaClient.send(ASSISTANT_EVENT_TOPIC.get(), key, assistant.toString());
  }

  /**
   * connector 实例停机/重启，通知 session-client 删除该节点的缓存.
   *
   * @see SessionClient#onShutdown
   */
  @Override
  public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) {
    if (event.getType() == PathChildrenCacheEvent.Type.CHILD_REMOVED) {
      if (client.getState() != CuratorFrameworkState.STOPPED) {
        final long timestamp = System.currentTimeMillis();

        final ChildData data = event.getData();
        final String path = data.getPath();

        final String nodeId = path.substring(path.lastIndexOf('/') + 1);

        final RedisExecutor<RSet<String>> nodeExecutor = RedissonOperation.NODE_ASSISTANT
            .executor(nodeId);

        // 忽略不是 connector 的停机事件
        if (!nodeExecutor.get(RSet::isExists, "isExists")) {
          return;
        }

        vertx.executeBlocking(promise -> {

          final RedisExecutor<RBucket<String>> executor = RedissonOperation.SHUTDOWN_LOCK
              .executor(nodeId);

          final boolean holdLock = executor.get(lock -> lock.compareAndSet(null, ""), "lock");

          // 发送停机消息给 session-client
          if (holdLock) {
            executor.run(bucket -> bucket.expireAsync(Duration.ofMinutes(5L)), "expireAsync");
            log.debug("received zk_node removed path {}", path);

            final Assistant shutdownEvent = Assistant.builder()
                .nodeId(nodeId)
                .action("SHUTDOWN")
                .timestamp(timestamp)
                .build();
            sendMessage(shutdownEvent);
          }
          promise.complete();
        })
            .onFailure(throwable -> log.error("process shutdown error", throwable));

        vertx.executeBlocking(promise -> {
          removeAssistantSessionOnNode(nodeExecutor, nodeId, timestamp);
          promise.complete();
        })
            .onFailure(throwable -> log.error("processAssistantSession failed", throwable));

      }
    }
  }

  /**
   * 停机/重启，删除该节点的客服 session.
   */
  private void removeAssistantSessionOnNode(
      RedisExecutor<RSet<String>> nodeSetExecutor,
      String nodeId,
      long shutdownTimestamp) {

    final int size = nodeSetExecutor.get(RSet::size, "size");
    final LongAdder completedCounter = new LongAdder();

    for (int i = 0; i < size; i++) {
      vertx.executeBlocking(promise -> {
        String address = null;
        try {
          // https://redis.io/commands/spop/
          // 利用 redis spop 随机移出一个元素的特性，多进程、多线程处理
          address = nodeSetExecutor.get(RSet::removeRandom, "removeRandom");
          if (address == null) {
            if (nodeSetExecutor.get(RObject::delete, "delete")) {
              log.info("connector shutdown process completed {}", completedCounter);
            }
            return;
          }

          final List<String> sidAndAid = DOT_SPLITTER.splitToList(address);

          final Assistant assistant = Assistant.builder()
              .storeId(sidAndAid.get(0))
              .assistantId(sidAndAid.get(1))
              .nodeId(nodeId)
              .action("LOGOUT")
              .timestamp(shutdownTimestamp)
              .build();

          updateAssistantSession(assistant, false);
          completedCounter.increment();
          promise.complete();
        } catch (Exception e) {
          log.error("process shutdown assistant failed, {} {}", address, nodeId, e);
          promise.fail(e);
        }
      });
    }
  }

}
